package com.gitee.huanminabc.utils_tools.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Zookeeper 监听器管理器
 * 专门负责管理各种类型的监听器，包括可持续监听器
 *
 * @author sby
 * @since 1.0.0
 */
@Slf4j
public class ZookeeperWatcherManager {

    private final ZooKeeper zooKeeper;
    private final ZookeeperUtil zkUtil;
    
    // 存储可持续监听器的映射（使用复合 key 避免路径冲突）
    private final Map<String, PersistentWatcher> persistentWatchers = new ConcurrentHashMap<>();
    private final Map<String, PersistentChildrenWatcher> persistentChildrenWatchers = new ConcurrentHashMap<>();
    
    // 重试管理
    private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(
        Math.max(2, Runtime.getRuntime().availableProcessors() / 2), // 修复：根据CPU核心数配置线程池
        r -> {
            Thread t = new Thread(r, "ZookeeperRetry-" + System.currentTimeMillis());
            t.setDaemon(true); // 设置为守护线程，避免阻止JVM退出
            return t;
        }
    );
    private final Map<String, AtomicInteger> retryCounts = new ConcurrentHashMap<>();
    private static final int MAX_RETRY_COUNT = 3;
    private static final long RETRY_DELAY_MS = 1000;
    private static final String RETRY_KEY_FORMAT = "%s:%s";
    
    // 生成复合 key 的辅助方法
    private String generateNodeWatcherKey(String path) {
        return "NODE:" + path;
    }
    
    private String generateChildrenWatcherKey(String path) {
        return "CHILDREN:" + path;
    }
    
    /**
     * 提取路径的辅助方法
     *
     * @param key 复合 key
     * @return 提取的路径，如果格式不正确则返回 null
     */
    private String extractPathFromKey(String key) {
        if (key == null || key.isEmpty()) {
            return null;
        }
        
        if (key.startsWith("NODE:")) {
            String path = key.substring(5);
            return path.isEmpty() ? null : path;
        } else if (key.startsWith("CHILDREN:")) {
            String path = key.substring(9);
            return path.isEmpty() ? null : path;
        }
        
        // 如果不是预期的格式，记录警告并返回 null
        log.warn("无效的监听器 key 格式: {}", key);
        return null;
    }
    
    /**
     * 构造函数
     *
     * @param zooKeeper ZooKeeper 实例
     * @param zkUtil ZookeeperUtil 实例
     */
    public ZookeeperWatcherManager(ZooKeeper zooKeeper, ZookeeperUtil zkUtil) {
        this.zooKeeper = zooKeeper;
        this.zkUtil = zkUtil;
    }
    
    /**
     * 可持续节点监听器
     */
    public static class PersistentWatcher implements Watcher {
        private final String path;
        private final Watcher userWatcher;
        private final ZookeeperWatcherManager manager;
        private final String data;
        
        public PersistentWatcher(String path, Watcher userWatcher, ZookeeperWatcherManager manager, String data) {
            this.path = path;
            this.userWatcher = userWatcher;
            this.manager = manager;
            this.data = data;
        }
        
        @Override
        public void process(WatchedEvent event) {
            try {
                // 1. 先重新注册监听器，确保可持续性（关键修复）
                switch (event.getType()) {
                    case NodeCreated:
                    case NodeDataChanged:
                    case NodeDeleted:
                        manager.reRegisterNodeWatcher(path, this, data);
                        break;
                    default:
                        break;
                }
                
                // 2. 再执行用户的监听器逻辑
                if (userWatcher != null) {
                    userWatcher.process(event);
                }
                
            } catch (Exception e) {
                log.error("可持续监听器处理事件失败: {}, 事件: {}", path, event, e);
                // 修复：不再异常后重新注册，避免死循环风险
                // 让系统自然恢复，避免无限循环
            }
        }
    }
    
    /**
     * 可持续子节点监听器
     */
    public static class PersistentChildrenWatcher implements Watcher {
        private final String path;
        private final Watcher userWatcher;
        private final ZookeeperWatcherManager manager;
        private final String data;
        
        public PersistentChildrenWatcher(String path, Watcher userWatcher, ZookeeperWatcherManager manager, String data) {
            this.path = path;
            this.userWatcher = userWatcher;
            this.manager = manager;
            this.data = data;
        }
        
        @Override
        public void process(WatchedEvent event) {
            try {
                // 1. 先重新注册监听器，确保可持续性（关键修复）
                switch (event.getType()) {
                    case NodeCreated:
                    case NodeDeleted:
                    case NodeChildrenChanged:
                        manager.reRegisterChildrenWatcher(path, this, data);
                        break;
                    default:
                        break;
                }
                
                // 2. 再执行用户的监听器逻辑
                if (userWatcher != null) {
                    userWatcher.process(event);
                }
                
            } catch (Exception e) {
                log.error("可持续子节点监听器处理事件失败: {}, 事件: {}", path, event, e);
                // 修复：不再异常后重新注册，避免死循环风险
                // 让系统自然恢复，避免无限循环
            }
        }
    }
    
    // ==================== 可持续监听方法 ====================
    
    /**
     * 监听节点变化（可持续监听）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @param data     节点数据（如果节点不存在时创建）
     * @return 是否添加成功
     */
    public boolean watchNode(String path, Watcher userWatcher, String data) {
        // 参数验证
        if (path == null || path.isEmpty()) {
            log.error("节点路径不能为空");
            return false;
        }
        if (userWatcher == null) {
            log.error("用户监听器不能为空: {}", path);
            return false;
        }
        // 修复：添加路径格式验证，防止恶意路径
        if (!path.startsWith("/")) {
            log.error("节点路径必须以 / 开头: {}", path);
            return false;
        }
        
        try {
            PersistentWatcher persistentWatcher = new PersistentWatcher(path, userWatcher, this, data);
            persistentWatchers.put(generateNodeWatcherKey(path), persistentWatcher);
            
            // 检查节点是否存在
            Stat stat = zooKeeper.exists(path, persistentWatcher);
            if (stat == null) {
                log.info("节点不存在，自动创建节点: {}", path);
                // 节点不存在，自动创建
                if (zkUtil.createNodeRecursive(path, data)) {
                    log.info("节点创建成功，可持续监听器添加成功: {}", path);
                    return true;
                } else {
                    log.error("节点创建失败，无法添加可持续监听器: {}", path);
                    persistentWatchers.remove(generateNodeWatcherKey(path));
                    return false;
                }
            }
            
            log.info("可持续节点监听器添加成功: {}", path);
            return true;
            
        } catch (Exception e) {
            log.error("添加可持续节点监听器失败: {}", path, e);
            persistentWatchers.remove(generateNodeWatcherKey(path));
            return false;
        }
    }
    
    /**
     * 监听子节点变化（可持续监听）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @param data     节点数据（如果节点不存在时创建）
     * @return 是否添加成功
     */
    public boolean watchChildren(String path, Watcher userWatcher, String data) {
        // 参数验证
        if (path == null || path.isEmpty()) {
            log.error("节点路径不能为空");
            return false;
        }
        if (userWatcher == null) {
            log.error("用户监听器不能为空: {}", path);
            return false;
        }
        // 修复：添加路径格式验证，防止恶意路径
        if (!path.startsWith("/")) {
            log.error("子节点路径必须以 / 开头: {}", path);
            return false;
        }
        
        try {
            PersistentChildrenWatcher persistentWatcher = new PersistentChildrenWatcher(path, userWatcher, this, data);
            persistentChildrenWatchers.put(generateChildrenWatcherKey(path), persistentWatcher);
            
            // 获取子节点列表并添加监听器
            List<String> children = zooKeeper.getChildren(path, persistentWatcher);
            log.info("可持续子节点监听器添加成功: {}, 当前子节点数量: {}", path, children.size());
            return true;
            
        } catch (KeeperException.NoNodeException e) {
            log.info("节点不存在，自动创建节点: {}", path);
            // 节点不存在，自动创建
            if (zkUtil.createNodeRecursive(path, data)) {
                try {
                    PersistentChildrenWatcher persistentWatcher = persistentChildrenWatchers.get(generateChildrenWatcherKey(path));
                    if (persistentWatcher != null) {
                        // 重新添加监听器
                        List<String> children = zooKeeper.getChildren(path, persistentWatcher);
                        log.info("节点创建成功，可持续子节点监听器添加成功: {}, 当前子节点数量: {}", path, children.size());
                        return true;
                    } else {
                        log.error("可持续监听器丢失: {}", path);
                        return false;
                    }
                } catch (Exception ex) {
                    log.error("重新添加可持续子节点监听器失败: {}", path, ex);
                    persistentChildrenWatchers.remove(generateChildrenWatcherKey(path));
                    return false;
                }
            } else {
                log.error("节点创建失败，无法添加可持续子节点监听器: {}", path);
                persistentChildrenWatchers.remove(generateChildrenWatcherKey(path));
                return false;
            }
        } catch (Exception e) {
            log.error("添加可持续子节点监听器失败: {}", path, e);
            persistentChildrenWatchers.remove(generateChildrenWatcherKey(path));
            return false;
        }
    }
    
    /**
     * 移除可持续节点监听器
     *
     * @param path 节点路径
     * @return 是否移除成功
     */
    public boolean removePersistentNodeWatcher(String path) {
        PersistentWatcher watcher = persistentWatchers.remove(generateNodeWatcherKey(path));
        if (watcher != null) {
            // 注意：Zookeeper 没有直接的监听器移除方法
            // 我们只能从本地映射中移除，下次事件触发时不会重新注册
            // 监听器会在 Zookeeper 端自然失效（因为不再重新注册）
            log.info("可持续节点监听器已从本地移除: {}", path);
            return true;
        }
        return false;
    }
    
    /**
     * 移除可持续子节点监听器
     *
     * @param path 节点路径
     * @return 是否移除成功
     */
    public boolean removePersistentChildrenWatcher(String path) {
        PersistentChildrenWatcher watcher = persistentChildrenWatchers.remove(generateChildrenWatcherKey(path));
        if (watcher != null) {
            // 注意：Zookeeper 没有直接的监听器移除方法
            // 我们只能从本地映射中移除，下次事件触发时不会重新注册
            // 监听器会在 Zookeeper 端自然失效（因为不再重新注册）
            log.info("可持续子节点监听器已从本地移除: {}", path);
            return true;
        }
        return false;
    }
    
    /**
     * 移除所有可持续监听器
     */
    public void removeAllPersistentWatchers() {
        try {
            // 获取所有路径的副本，避免并发修改异常
            List<String> nodePaths = new ArrayList<>(persistentWatchers.keySet());
            List<String> childrenPaths = new ArrayList<>(persistentChildrenWatchers.keySet());
            
            // 移除节点监听器
            for (String path : nodePaths) {
                removePersistentNodeWatcher(extractPathFromKey(path));
            }
            
            // 移除子节点监听器
            for (String path : childrenPaths) {
                removePersistentChildrenWatcher(extractPathFromKey(path));
            }
            
            log.info("所有可持续监听器已移除，共移除 {} 个", nodePaths.size() + childrenPaths.size());
        } catch (Exception e) {
            log.error("移除所有监听器时发生异常", e);
        }
    }
    
    /**
     * 获取当前可持续监听器数量
     *
     * @return 监听器数量
     */
    public int getPersistentWatcherCount() {
        return persistentWatchers.size() + persistentChildrenWatchers.size();
    }
    
    /**
     * 获取节点监听器数量
     *
     * @return 节点监听器数量
     */
    public int getNodeWatcherCount() {
        return persistentWatchers.size();
    }
    
    /**
     * 获取子节点监听器数量
     *
     * @return 子节点监听器数量
     */
    public int getChildrenWatcherCount() {
        return persistentChildrenWatchers.size();
    }
    
    /**
     * 检查特定路径是否有监听器
     *
     * @param path 节点路径
     * @return 是否有监听器
     */
    public boolean hasWatcher(String path) {
        if (path == null || path.isEmpty()) {
            return false;
        }
        return persistentWatchers.containsKey(generateNodeWatcherKey(path)) || 
               persistentChildrenWatchers.containsKey(generateChildrenWatcherKey(path));
    }
    
    /**
     * 获取所有监听器路径
     *
     * @return 监听器路径列表
     */
    public List<String> getAllWatcherPaths() {
        List<String> paths = new ArrayList<>();
        // 使用同步块确保线程安全
        synchronized (persistentWatchers) {
            for (String key : persistentWatchers.keySet()) {
                paths.add(extractPathFromKey(key));
            }
        }
        synchronized (persistentChildrenWatchers) {
            for (String key : persistentChildrenWatchers.keySet()) {
                paths.add(extractPathFromKey(key));
            }
        }
        return Collections.unmodifiableList(paths);
    }
    
    /**
     * 获取节点监听器路径列表
     *
     * @return 节点监听器路径列表
     */
    public List<String> getNodeWatcherPaths() {
        List<String> paths = new ArrayList<>();
        synchronized (persistentWatchers) {
            for (String key : persistentWatchers.keySet()) {
                paths.add(extractPathFromKey(key));
            }
        }
        return Collections.unmodifiableList(paths);
    }
    
    /**
     * 获取子节点监听器路径列表
     *
     * @return 子节点监听器路径列表
     */
    public List<String> getChildrenWatcherPaths() {
        List<String> paths = new ArrayList<>();
        synchronized (persistentChildrenWatchers) {
            for (String key : persistentChildrenWatchers.keySet()) {
                paths.add(extractPathFromKey(key));
            }
        }
        return Collections.unmodifiableList(paths);
    }
    
    /**
     * 暂停监听器（临时禁用，不删除）
     *
     * @param path 节点路径
     * @return 是否暂停成功
     */
    public boolean pauseWatcher(String path) {
        // 这里可以实现暂停逻辑，比如设置一个暂停标志
        log.info("监听器暂停功能待实现: {}", path);
        return true;
    }
    
    /**
     * 恢复监听器
     *
     * @param path 节点路径
     * @return 是否恢复成功
     */
    public boolean resumeWatcher(String path) {
        // 这里可以实现恢复逻辑
        log.info("监听器恢复功能待实现: {}", path);
        return true;
    }
    
    /**
     * 重新注册节点监听器（内部方法）
     *
     * @param path 节点路径
     * @param watcher 监听器
     * @param data 节点数据
     */
    void reRegisterNodeWatcher(String path, PersistentWatcher watcher, String data) {
        try {
            if (zooKeeper.getState() == ZooKeeper.States.CONNECTED) {
                zooKeeper.exists(path, watcher);
                log.debug("节点监听器重新注册成功: {}", path);
            } else {
                log.warn("Zookeeper 未连接，无法重新注册监听器: {}, 当前状态: {}", 
                    path, getConnectionStateDescription());
                // 可以在这里添加重连逻辑
            }
        } catch (KeeperException.NoNodeException e) {
            log.warn("节点不存在，无法重新注册监听器: {}", path);
            // 节点不存在时，不进行重试
        } catch (KeeperException.ConnectionLossException e) {
            log.warn("连接丢失，无法重新注册监听器: {}", path);
            // 连接丢失时，安排重试
            scheduleRetry(path, watcher, data, "node");
        } catch (KeeperException.SessionExpiredException e) {
            log.error("会话过期，无法重新注册监听器: {}", path);
            // 会话过期时，不进行重试，需要重新连接
        } catch (Exception e) {
            log.error("重新注册节点监听器失败: {}, 错误类型: {}", path, e.getClass().getSimpleName(), e);
            // 其他异常时，安排重试
            scheduleRetry(path, watcher, data, "node");
        }
    }
    
    /**
     * 重新注册子节点监听器（内部方法）
     *
     * @param path 节点路径
     * @param watcher 监听器
     * @param data 节点数据
     */
    void reRegisterChildrenWatcher(String path, PersistentChildrenWatcher watcher, String data) {
        try {
            if (zooKeeper.getState() == ZooKeeper.States.CONNECTED) {
                zooKeeper.getChildren(path, watcher);
                log.debug("子节点监听器重新注册成功: {}", path);
            } else {
                log.warn("Zookeeper 未连接，无法重新注册监听器: {}, 当前状态: {}", 
                    path, getConnectionStateDescription());
                // 可以在这里添加重连逻辑
            }
        } catch (KeeperException.NoNodeException e) {
            log.warn("节点不存在，无法重新注册监听器: {}", path);
            // 节点不存在时，不进行重试
        } catch (KeeperException.ConnectionLossException e) {
            log.warn("连接丢失，无法重新注册监听器: {}", path);
            // 连接丢失时，安排重试
            scheduleRetry(path, watcher, data, "children");
        } catch (KeeperException.SessionExpiredException e) {
            log.error("会话过期，无法重新注册监听器: {}", path);
            // 会话过期时，不进行重试，需要重新连接
        } catch (Exception e) {
            log.error("重新注册子节点监听器失败: {}, 错误类型: {}", path, e.getClass().getSimpleName(), e);
            // 其他异常时，安排重试
            scheduleRetry(path, watcher, data, "children");
        }
    }
    
    /**
     * 安排重试重新注册（改进版本：使用线程池和重试次数限制）
     *
     * @param path 节点路径
     * @param watcher 监听器
     * @param data 节点数据
     * @param type 监听器类型
     */
    private void scheduleRetry(String path, Watcher watcher, String data, String type) {
        String retryKey = String.format(RETRY_KEY_FORMAT, path, type); // 修复：使用格式化字符串，减少对象创建
        AtomicInteger retryCount = retryCounts.computeIfAbsent(retryKey, k -> new AtomicInteger(0));
        
        int currentRetry = retryCount.incrementAndGet();
        if (currentRetry > MAX_RETRY_COUNT) {
            log.error("重试次数已达上限，停止重试: {}, 类型: {}, 路径: {}", currentRetry, type, path);
            retryCounts.remove(retryKey);
            return;
        }
        
        long delay = RETRY_DELAY_MS * currentRetry; // 指数退避
        log.debug("安排重试重新注册监听器: {}, 类型: {}, 第 {} 次重试, 延迟 {}ms", path, type, currentRetry, delay); // 修复：改为DEBUG级别，减少生产环境日志量
        
        retryExecutor.schedule(() -> {
            try {
                if (zooKeeper.getState() == ZooKeeper.States.CONNECTED) {
                    if ("node".equals(type)) {
                        zooKeeper.exists(path, watcher);
                        log.debug("重试重新注册节点监听器成功: {}, 第 {} 次重试", path, currentRetry); // 修复：改为DEBUG级别
                        retryCounts.remove(retryKey); // 成功后移除重试计数
                    } else if ("children".equals(type)) {
                        zooKeeper.getChildren(path, watcher);
                        log.debug("重试重新注册子节点监听器成功: {}, 第 {} 次重试", path, currentRetry); // 修复：改为DEBUG级别
                        retryCounts.remove(retryKey); // 成功后移除重试计数
                    }
                } else {
                    log.warn("Zookeeper 未连接，重试失败: {}, 类型: {}", path, type);
                }
            } catch (Exception e) {
                log.error("重试重新注册监听器失败: {}, 类型: {}, 第 {} 次重试", path, type, currentRetry, e);
                // 修复：不再递归调用，避免无限递归
                // 重试失败后，等待下次事件触发时再尝试重新注册
            }
        }, delay, TimeUnit.MILLISECONDS);
        
        // 修复：添加超时清理机制，防止重试计数无限累积
        retryExecutor.schedule(() -> {
            if (retryCounts.remove(retryKey) != null) {
                log.debug("清理过期的重试计数: {}", retryKey);
            }
        }, RETRY_DELAY_MS * MAX_RETRY_COUNT + 5000, TimeUnit.MILLISECONDS);
    }
    
    // ==================== 数据变化监听方法 ====================
    
    /**
     * 监听节点数据变化（可持续监听）
     *
     * @param path 节点路径
     * @param dataChangeHandler 数据变化处理器
     * @param data 初始节点数据（如果节点不存在时创建）
     * @return 是否添加成功
     */
    public boolean watchDataChange(String path, DataChangeHandler dataChangeHandler, String data) {
        String key = generateNodeWatcherKey(path);
        try {
            // 创建可持续的数据变化监听器（关键修复：使用 PersistentWatcher）
            PersistentWatcher persistentWatcher = new PersistentWatcher(path, 
                dataChangeHandler.createWatcher(this), this, data);
            
            // 存储到可持续监听器映射中
            persistentWatchers.put(key, persistentWatcher);
            
            // 检查节点是否存在并添加监听器
            Stat stat = zooKeeper.exists(path, persistentWatcher);
            if (stat == null) {
                log.info("节点不存在，自动创建节点: {}", path);
                // 节点不存在，自动创建
                if (zkUtil.createNodeRecursive(path, data)) {
                    log.info("节点创建成功，数据变化监听器添加成功: {}", path);
                    return true;
                } else {
                    log.error("节点创建失败，无法添加数据变化监听器: {}", path);
                    persistentWatchers.remove(key);
                    return false;
                }
            }
            
            log.info("数据变化监听器添加成功: {}", path);
            return true;
            
        } catch (Exception e) {
            log.error("添加数据变化监听器失败: {}", path, e);
            // 修复：异常时清理已添加的监听器，防止内存泄漏
            persistentWatchers.remove(key);
            return false;
        }
    }
    
    /**
     * 监听节点数据变化（可持续监听，自动创建空节点）
     *
     * @param path 节点路径
     * @param dataChangeHandler 数据变化处理器
     * @return 是否添加成功
     */
    public boolean watchDataChange(String path, DataChangeHandler dataChangeHandler) {
        return watchDataChange(path, dataChangeHandler, "");
    }
    
    /**
     * 监听节点数据变化（可持续监听，带初始数据）
     *
     * @param path 节点路径
     * @param dataChangeHandler 数据变化处理器
     * @param data 初始节点数据
     * @param createIfNotExists 如果节点不存在是否自动创建
     * @return 是否添加成功
     */
    public boolean watchDataChange(String path, DataChangeHandler dataChangeHandler, String data, boolean createIfNotExists) {
        if (!createIfNotExists) {
            // 如果不需要自动创建，直接添加监听器
            return watchNode(path, dataChangeHandler.createWatcher(this), data);
        }
        return watchDataChange(path, dataChangeHandler, data);
    }
    
    /**
     * 数据变化处理器接口
     */
    public interface DataChangeHandler {
        
        /**
         * 数据变化回调
         *
         * @param path 节点路径
         * @param oldData 旧数据
         * @param newData 新数据
         */
        void onDataChanged(String path, String oldData, String newData);
        
        /**
         * 获取上次的数据
         *
         * @return 上次的数据
         */
        default String getLastData() {
            return null;
        }
        
        /**
         * 设置数据缓存
         *
         * @param data 数据
         */
        default void setLastData(String data) {
            // 默认实现为空，子类可以重写
        }
        
        /**
         * 创建监听器（内部使用）
         *
         * @param manager ZookeeperWatcherManager 实例
         * @return 监听器
         */
        default Watcher createWatcher(ZookeeperWatcherManager manager) {
            // 获取当前实例的路径（通过反射或其他方式）
            final String currentPath = getCurrentPath();
            return new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeDataChanged) {
                        try {
                            // 修复：使用当前实例的路径
                            String newData = manager.zkUtil.getNodeData(currentPath);
                            String oldData = getLastData();
                            onDataChanged(currentPath, oldData, newData);
                            setLastData(newData);
                        } catch (Exception e) {
                            log.error("处理数据变化事件失败: {}", currentPath, e);
                        }
                    }
                }
            };
        }
        
        /**
         * 获取当前路径（子类需要实现）
         *
         * @return 当前路径
         */
        default String getCurrentPath() {
            // 默认实现，子类应该重写
            return null;
        }
    }
    
    /**
     * 简单数据变化处理器（带数据缓存）
     */
    public static class SimpleDataChangeHandler implements DataChangeHandler {
        private String lastData;
        private final String path;
        
        public SimpleDataChangeHandler(String path) {
            this.path = path;
        }
        
        @Override
        public void onDataChanged(String path, String oldData, String newData) {
            log.info("节点数据变化: {} -> {}", oldData, newData);
        }
        
        @Override
        public String getLastData() {
            return lastData;
        }
        
        @Override
        public void setLastData(String data) {
            this.lastData = data;
        }

        @Override
        public String getCurrentPath() {
            return path;
        }
    }
    
    /**
     * 带业务逻辑的数据变化处理器
     */
    public static class BusinessDataChangeHandler implements DataChangeHandler {
        private String lastData;
        private final String path;
        private final Runnable onDataChanged;
        
        public BusinessDataChangeHandler(String path, Runnable onDataChanged) {
            this.path = path;
            this.onDataChanged = onDataChanged;
        }
        
        @Override
        public void onDataChanged(String path, String oldData, String newData) {
            log.info("业务数据变化: {} -> {}", oldData, newData);
            if (onDataChanged != null) {
                try {
                    onDataChanged.run();
                } catch (Exception e) {
                    log.error("执行业务逻辑失败", e);
                }
            }
        }
        
        @Override
        public String getLastData() {
            return lastData;
        }
        
        @Override
        public void setLastData(String data) {
            this.lastData = data;
        }

        @Override
        public String getCurrentPath() {
            return path;
        }
    }
    
    // ==================== 便捷方法 ====================
    
    /**
     * 添加可持续节点监听器（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @return 是否添加成功
     */
    public boolean watchNode(String path, Watcher userWatcher) {
        return watchNode(path, userWatcher, "");
    }
    
    /**
     * 添加可持续子节点监听器（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @return 是否添加成功
     */
    public boolean watchChildren(String path, Watcher userWatcher) {
        return watchChildren(path, userWatcher, "");
    }
    
    /**
     * 添加可持续节点监听器并自动创建节点（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @param data     节点数据
     * @return 是否添加成功
     */
    public boolean watchPersistentAndCreate(String path, Watcher userWatcher, String data) {
        return watchNode(path, userWatcher, data);
    }
    
    /**
     * 添加可持续子节点监听器并自动创建节点（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @param data     节点数据
     * @return 是否添加成功
     */
    public boolean watchPersistentChildrenAndCreate(String path, Watcher userWatcher, String data) {
        return watchChildren(path, userWatcher, data);
    }
    
    /**
     * 添加可持续节点监听器并自动创建空节点（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @return 是否添加成功
     */
    public boolean watchPersistentAndCreateEmpty(String path, Watcher userWatcher) {
        return watchNode(path, userWatcher, "");
    }
    
    /**
     * 添加可持续子节点监听器并自动创建空节点（便捷方法）
     *
     * @param path     节点路径
     * @param userWatcher 用户自定义监听器
     * @return 是否添加成功
     */
    public boolean watchPersistentChildrenAndCreateEmpty(String path, Watcher userWatcher) {
        return watchChildren(path, userWatcher, "");
    }
    
    // ==================== 数据变化监听便捷方法 ====================
    
    /**
     * 监听数据变化并执行回调（便捷方法）
     *
     * @param path 节点路径
     * @param callback 数据变化回调
     * @param initialData 初始数据
     * @return 是否添加成功
     */
    public boolean watchDataChangeWithCallback(String path, DataChangeCallback callback, String initialData) {
        final String finalPath = path; // 使用 final 变量避免遮蔽
        DataChangeHandler handler = new DataChangeHandler() {
            @Override
            public void onDataChanged(String eventPath, String oldData, String newData) {
                // 修复：使用传入的 path 参数，保持一致性
                callback.onDataChanged(eventPath, oldData, newData);
            }
            
            @Override
            public String getCurrentPath() {
                return finalPath; // 返回正确的路径
            }
        };
        return watchDataChange(finalPath, handler, initialData);
    }
    
    /**
     * 监听数据变化并执行回调（便捷方法，自动创建空节点）
     *
     * @param path 节点路径
     * @param callback 数据变化回调
     * @return 是否添加成功
     */
    public boolean watchDataChangeWithCallback(String path, DataChangeCallback callback) {
        return watchDataChangeWithCallback(path, callback, "");
    }
    
    /**
     * 数据变化回调接口
     */
    @FunctionalInterface
    public interface DataChangeCallback {
        /**
         * 数据变化回调
         *
         * @param path 节点路径
         * @param oldData 旧数据
         * @param newData 新数据
         */
        void onDataChanged(String path, String oldData, String newData);
    }

    /**
     * 检查 Zookeeper 连接状态
     *
     * @return 是否已连接
     */
    public boolean isConnected() {
        return zooKeeper != null && zooKeeper.getState() == ZooKeeper.States.CONNECTED;
    }
    
    /**
     * 获取 Zookeeper 连接状态
     *
     * @return 连接状态
     */
    public ZooKeeper.States getConnectionState() {
        return zooKeeper != null ? zooKeeper.getState() : null;
    }
    
    /**
     * 检查连接是否可用（包括连接和认证状态）
     *
     * @return 连接是否可用
     */
    public boolean isConnectionAvailable() {
        if (zooKeeper == null) {
            return false;
        }
        
        ZooKeeper.States state = zooKeeper.getState();
        return state == ZooKeeper.States.CONNECTED || 
               state == ZooKeeper.States.CONNECTEDREADONLY;
    }
    
    /**
     * 获取连接状态描述
     *
     * @return 连接状态描述
     */
    public String getConnectionStateDescription() {
        if (zooKeeper == null) {
            return "未初始化";
        }
        
        ZooKeeper.States state = zooKeeper.getState();
        switch (state) {
            case CONNECTING:
                return "连接中";
            case CONNECTED:
                return "已连接";
            case CONNECTEDREADONLY:
                return "已连接（只读）";
            case CLOSED:
                return "已关闭";
            case AUTH_FAILED:
                return "认证失败";
            default:
                return "未知状态: " + state;
        }
    }
    
    /**
     * 重新注册所有监听器（用于重连后恢复监听器）
     *
     * @return 重新注册成功的监听器数量
     */
    public int reRegisterAllWatchers() {
        if (!isConnected()) {
            log.warn("Zookeeper 未连接，无法重新注册监听器");
            return 0;
        }
        
        int successCount = 0;
        int totalCount = 0;
        int consecutiveFailures = 0; // 修复：添加连续失败计数，防止系统崩溃
        
        try {
            // 重新注册节点监听器
            for (Map.Entry<String, PersistentWatcher> entry : persistentWatchers.entrySet()) {
                // 修复：如果连续失败次数过多，停止处理，防止系统崩溃
                if (consecutiveFailures > 10) {
                    log.error("连续失败次数过多，停止重新注册监听器，防止系统崩溃");
                    break;
                }
                
                String path = extractPathFromKey(entry.getKey());
                if (path != null) { // 添加空值检查
                    PersistentWatcher watcher = entry.getValue();
                    try {
                        zooKeeper.exists(path, watcher);
                        successCount++;
                        consecutiveFailures = 0; // 成功后重置计数
                        log.debug("重新注册节点监听器成功: {}", path);
                    } catch (Exception e) {
                        consecutiveFailures++;
                        log.error("重新注册节点监听器失败: {}, 连续失败次数: {}", path, consecutiveFailures, e);
                    }
                    totalCount++;
                } else {
                    log.warn("无效的节点监听器 key: {}", entry.getKey());
                }
            }
            
            // 重新注册子节点监听器
            for (Map.Entry<String, PersistentChildrenWatcher> entry : persistentChildrenWatchers.entrySet()) {
                // 修复：如果连续失败次数过多，停止处理，防止系统崩溃
                if (consecutiveFailures > 10) {
                    log.error("连续失败次数过多，停止重新注册监听器，防止系统崩溃");
                    break;
                }
                
                String path = extractPathFromKey(entry.getKey());
                if (path != null) { // 添加空值检查
                    PersistentChildrenWatcher watcher = entry.getValue();
                    try {
                        zooKeeper.getChildren(path, watcher);
                        successCount++;
                        consecutiveFailures = 0; // 成功后重置计数
                        log.debug("重新注册子节点监听器成功: {}", path);
                    } catch (Exception e) {
                        consecutiveFailures++;
                        log.error("重新注册子节点监听器失败: {}, 连续失败次数: {}", path, consecutiveFailures, e);
                    }
                    totalCount++;
                } else {
                    log.warn("无效的子节点监听器 key: {}", entry.getKey());
                }
            }
            
            log.info("重新注册监听器完成，成功: {}/{}", successCount, totalCount);
            
        } catch (Exception e) {
            log.error("重新注册所有监听器时发生严重异常", e);
        }
        
        return successCount;
    }
    
    /**
     * 清理资源（关闭线程池等）
     */
    public void cleanup() {
        try {
            // 关闭重试线程池
            if (!retryExecutor.isShutdown()) {
                retryExecutor.shutdown();
                try {
                    if (!retryExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                        log.warn("重试线程池未能在5秒内关闭，强制关闭");
                        retryExecutor.shutdownNow();
                        // 再次等待，确保线程池完全关闭
                        if (!retryExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                            log.error("重试线程池无法完全关闭");
                        }
                    }
                } catch (InterruptedException e) {
                    log.warn("等待线程池关闭时被中断");
                    retryExecutor.shutdownNow();
                    Thread.currentThread().interrupt(); // 恢复中断状态
                }
            }
            
            // 清理重试计数和监听器映射
            retryCounts.clear();
            persistentWatchers.clear();
            persistentChildrenWatchers.clear();
            
            log.info("ZookeeperWatcherManager 资源清理完成");
        } catch (Exception e) {
            log.error("清理资源时发生异常", e);
        }
    }
    
    /**
     * 获取重试统计信息
     *
     * @return 重试统计信息
     */
    public String getRetryStatistics() {
        StringBuilder sb = new StringBuilder();
        sb.append("重试统计信息:\n");
        sb.append("当前重试任务数: ").append(retryCounts.size()).append("\n");
        sb.append("线程池状态: ").append(retryExecutor.isShutdown() ? "已关闭" : "运行中").append("\n");
        
        if (!retryCounts.isEmpty()) {
            sb.append("重试详情:\n");
            retryCounts.forEach((key, count) -> {
                sb.append("  ").append(key).append(": ").append(count.get()).append(" 次\n");
            });
        }
        
        return sb.toString();
    }
}
